Scraper¶
In [1]:
import sys
import urllib.request
from urllib.request import Request, urlopen
from urllib.request import URLError, HTTPError
from urllib.parse import quote
import http.client
from http.client import IncompleteRead, BadStatusLine
http.client._MAXHEADERS = 1000
import time
import os
import ssl
import datetime
import json
import socket
from fake_useragent import UserAgent
import chromedriver_binary
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
import webdrivermanager
from keras import applications
In [2]:
class image_download:
"""
Class based on google-images-download module created by hardikvasa.
I applied few changes and remove unnecessary methods to make class smaller,
more suitable for my needs and supported with documentation.
"""
def __init__(self, arguments):
self.arguments = arguments
pass
# Downloading entire Web Document (Raw Page Content)
def download_page(self,url):
"""
Method downloading page as raw text.
:param str url - url to the page we want to download.
:returns:
- str text - raw text of HTML page.
"""
try:
ua = UserAgent()
headers = {'User-Agent':ua.chrome}
req = urllib.request.Request(url, headers=headers)
text = str(urllib.request.urlopen(req).read())
return text
except Exception as e:
print("Could not open URL. Please check your internet connection.")
sys.exit()
def download_extended_page(self, url):
"""
Method downloading whole page from top to the bottom. Useful
when page don't load to the bottom before scrolling.
:param str url - url to the page we want to download.
:returns:
- str source - raw text of HTML page.
"""
options = webdriver.ChromeOptions()
options.add_argument('--no-sandbox')
options.add_argument("--headless")
try:
browser = webdriver.Chrome(options=options)
except Exception as e:
print("Looks like we cannot locate the path the 'chromedriver' or google chrome browser is not "
"installed on your machine (exception: %s)" % e)
sys.exit()
browser.set_window_size(1024, 768)
# Open the link
browser.get(url)
time.sleep(1)
print("Getting you a lot of images. This may take a few moments...")
element = browser.find_element_by_tag_name("body")
# Scroll down
for i in range(30):
element.send_keys(Keys.PAGE_DOWN)
time.sleep(0.3)
try:
browser.find_element_by_id("smb").click()
for i in range(50):
element.send_keys(Keys.PAGE_DOWN)
time.sleep(0.3) # bot id protection
except:
for i in range(10):
element.send_keys(Keys.PAGE_DOWN)
time.sleep(0.3) # bot id protection
print("Reached end of Page.")
time.sleep(0.5)
source = browser.page_source #page source
#close the browser
browser.close()
return source
def format_object(self,my_object):
"""
Method creating new dictionary with more readable keys.
:param dict my_object - dictionary obtained as json.loads() result.
:returns:
- dict formatted_object - new more readable dict.
"""
formatted_object = {}
formatted_object['image_format'] = my_object['ity']
formatted_object['image_link'] = my_object['ou']
formatted_object['image_source'] = my_object['ru']
return formatted_object
# make directories
def create_directories(self, dir_name):
"""
Method creating new directory if not exist.
:param str dir_name - path to the root dictionary.
"""
path_name = dir_name
if not os.path.isdir(path_name):
os.makedirs(path_name)
return
def handle_exception(self, error):
"""
Method handling exceptions which occurred during downloading images.
:param error - error which occurred during downloading images.
:returns:
- str download_status - info about download status.
- str download_message - info about error.
- str return_image_name - empty info about image filename - because download failed.
- str absolute_path - empty path to the downloaded file - because download failed.
"""
download_status = 'fail'
download_message = "Error occurred ====> " + str(error) + '\t ...trying next one...'
return_image_name = ''
absolute_path = ''
return download_status, download_message, return_image_name, absolute_path
# Download Images
def download_image(self,image_url, image_format, dir_name, count):
"""
Method handling exceptions which occurred during downloading images.
:param str image_url - link to the image.
:param str image_format - format of the image.
:param str dir_name - path to the directory.
:param str count - index of image in particular class.
:returns:
- str download_status - info about download status.
- str download_message - info about downloaded file.
- str return_image_name - info about image filename.
- str absolute_path -path to the downloaded file.
"""
try:
ua = UserAgent()
req = Request(image_url, headers = {'User-Agent':ua.chrome})
try:
# timeout time to download an image
timeout = 10
response = urlopen(req, None, timeout)
data = response.read()
response.close()
extensions = [".jpg", ".jpeg", ".png", ".bmp"]
# keep everything after the last '/'
image_name = self.arguments['image_name'].replace(' ', '_')
if image_format == "" or not image_format or "." + image_format not in extensions:
download_status, download_message, return_image_name, absolute_path = self.handle_exception("Invalid or missing image format.")
return download_status, download_message, return_image_name, absolute_path
elif image_name.lower().find("." + image_format) < 0:
image_name = image_name + "." + image_format
else:
image_name = image_name[:image_name.lower().find("." + image_format) + (len(image_format) + 1)]
path = dir_name + "/" + str(count) + "." + image_name
files = os.listdir('data_food_classifier_MG')
if sum([1 if str(count) + "." + self.arguments['image_name'].replace(' ', '_') in s else 0 for s in files]):
count = max([int(s.split('.')[0]) for s in files if self.arguments['image_name'].replace(' ', '_') in s]) + 1
path = dir_name + "/" + str(count) + "." + image_name
try:
output_file = open(path, 'wb')
output_file.write(data)
output_file.close()
absolute_path = os.path.abspath(path)
except OSError as e:
download_status, download_message, return_image_name, absolute_path = self.handle_exception(e)
#return image name back to calling method to use it for thumbnail downloads
download_status = 'success'
download_message = "Completed Image ====> " + str(count) + "." + image_name
return_image_name = str(count) + "." + image_name
except (UnicodeEncodeError, URLError, BadStatusLine) as e:
download_status, download_message, return_image_name, absolute_path = self.handle_exception(e)
except (HTTPError, URLError, ssl.CertificateError, IOError, IncompleteRead) as e: # If there is any HTTPError
download_status, download_message, return_image_name, absolute_path = self.handle_exception(e)
return download_status, download_message, return_image_name, absolute_path
def _get_next_item(self,s):
"""
Method finding next image from the given raw page.
:param str s - page raw text.
:returns:
- str final_object - encoded json.
- str end_object - div ending the current image.
"""
start_line = s.find('rg_meta notranslate')
if start_line == -1: # If no links are found then give an error!
end_quote = 0
link = "no_links"
return link, end_quote
else:
start_line = s.find('class="rg_meta notranslate">')
start_object = s.find('{', start_line + 1)
end_object = s.find('</div>', start_object + 1)
object_raw = str(s[start_object:end_object])
try:
object_decode = bytes(object_raw, "utf-8").decode("unicode_escape")
final_object = json.loads(object_decode)
except:
final_object = ""
return final_object, end_object
# Getting all links with the help of '_images_get_next_image'
def _get_all_items(self, page, dir_name, limit, attempts=0):
"""
Method getting all links with help of the '_images_get_next_image'.
:param str page - page raw text.
:param str dir_name - path to the directory.
:param int limit - number of images to download.
:param int attempts - number of attempts to download image (max 3).
:returns:
- int errorCount - number of encountered errors.
"""
errorCount = 0
i = 0
count = 1
while count < limit+1:
final_object, end_content = self._get_next_item(page)
if final_object == "no_links":
break
elif final_object == "":
page = page[end_content:]
else:
#format the item for readability
final_object = self.format_object(final_object)
#download the images
download_status, download_message, return_image_name, absolute_path = self.download_image(final_object['image_link'],
final_object['image_format'],
dir_name,
count)
print(download_message)
if download_status == "success":
count += 1
final_object['image_filename'] = return_image_name
else:
errorCount += 1
page = page[end_content:]
i += 1
if count - 1 == 0 and attempts < 3:
print('Internet connection is lost or there is no images with given name. \t ...trying again')
time.sleep(10)
attempts += 1
self._get_all_items(page, dir_name, limit, attempts)
elif count < limit:
print("\n\nUnfortunately all " + str(
limit) + " could not be downloaded because some images were not downloadable. " + str(
count-1) + " is all we got for this search filter!")
return errorCount
def download_executor(self):
"""
Method starting download processes.
:returns:
- list errors - list of errors.
"""
errorCount = None
######Initialization and Validation of user arguments
if self.arguments['keywords']:
search_keyword = [str(item) for item in self.arguments['keywords'].split(',')]
# Setting limit on number of images to be downloaded
if self.arguments['limit']:
limit = int(self.arguments['limit'])
else:
limit = 100
total_errors = 0
i = 0
while i < len(search_keyword): # 3.for every main keyword
iteration = "\n" + "Item no.: " + str(i + 1) + " -->" + " Item name = " + (search_keyword[i])
print("Downloading images for: " + (search_keyword[i]) + " ...")
search_term = search_keyword[i]
dir_name = 'data_food_classifier_MG' #sub-directory
self.create_directories(dir_name) #create directories in OS
url = 'https://www.google.com/search?q=' + quote(
search_term.encode('utf-8')) + '&espv=2&biw=1366&bih=667&site=webhp&source=lnms&tbm=isch&tbs=&sa=X&ei=XosDVaCXD8TasATItgE&ved=0CAcQ_AUoAg'
if limit < 101:
raw_html = self.download_page(url) # download page
else:
raw_html = self.download_extended_page(url)
errorCount= self._get_all_items(raw_html,dir_name,limit) #get all image items and download images
#dumps into a json file
i += 1
total_errors = total_errors + errorCount
return total_errors
In [3]:
def scrap_images(keyword, limit, image_name):
arguments = {"keywords":keyword, "limit":limit, "image_name":image_name}
total_errors = 0
t0 = time.time() # start the timer
downloader = image_download(arguments)
errors = downloader.download_executor() #wrapping response in a variable just for consistency
total_errors = total_errors + errors
t1 = time.time() # stop the timer
total_time = t1 - t0 # Calculating the total time required to crawl, find and download all the links of 60,000 images
print("\nEverything downloaded!")
print("Total errors: " + str(total_errors))
print("Total time taken: " + str(total_time) + " Seconds")
In [4]:
scrap_images('fast food', 300, 'fast food')
In [5]:
scrap_images('hamburger, kebab, pizza, french fries, chicken popeyes, burito', 100, 'fast food')
In [6]:
scrap_images('healthy meal', 300, 'slow food')
In [7]:
scrap_images('vegetables, fruit, fit meal, walnuts dinner, fish dinner, green beans dinner', 100, 'slow food')
Image preparation¶
In [8]:
import matplotlib.pyplot as plt
import matplotlib.image as img
import numpy as np
from PIL import Image
%matplotlib inline
In [9]:
root_dir = 'data_food_classifier_MG'
root_listdir = os.listdir(root_dir)
Some images may have been saved with the wrong extension. In addition, we must have consistency of extensions, so png images must be converted to jpg files.
In [10]:
counter = 0
counter_2 = 0
for i in range(len(root_listdir)):
filename = root_dir + '/' + root_listdir[i]
try:
im = Image.open(filename)
if '.png' in filename:
im = im.convert('RGB')
os.remove(filename)
filename = filename.replace('png', 'jpg')
root_listdir[i] = root_listdir[i].replace('png', 'jpg')
im.save(filename)
im = Image.open(filename)
counter_2 += 1
im.verify() #I perform also verify, don't know if he sees other types o defects
im.close() #reload is necessary in my case
except (OSError, ValueError):
os.remove(filename)
root_listdir[i] = 0
counter += 1
print('%d files deleted due to OSError and ValueError.' %counter)
print('%d files converted from png to jpg.' %counter_2)
In [11]:
root_listdir = [i for i in root_listdir if i != 0]
list_of_images_indices = np.random.choice(len(root_listdir), 60)
fig, ax = plt.subplots(10, 6, frameon=False, figsize=(15, 25))
fig.suptitle('Random image from food class', fontsize=20)
row = 0
col = 0
ec = (0, .6, .1)
fc = (0, .7, .2)
for i in list_of_images_indices:
try:
image = Image.open(root_dir + '/' + root_listdir[i])
image = image.resize((120, 120), Image.ANTIALIAS)
class_name = root_listdir[i].split('.')[1].replace('_', ' ')
ax[row][col].imshow(image)
ax[row][col].text(0, -20, class_name, size=10, rotation=0, ha='left',
va='top', bbox=dict(boxstyle='round', ec=ec, fc=fc))
ax[row][col].set_xticks([])
ax[row][col].set_yticks([])
col += 1
if col % 6 == 0 and col*row != 54:
col = 0
row += 1
elif col*row == 54:
break
except (OSError, ValueError) as e:
print(e)
plt.tight_layout(rect=[0, 0.03, 1, 0.95])
Splitting images into train, test and valid sets.
In [12]:
import shutil
len_data = len(root_listdir)
train_set = 0.8
train_examples = round(len_data*train_set)
valid_examples = round((len_data - train_examples)*0.2)
test_examples = len_data - train_examples - valid_examples
labels = np.asarray([1 if 'fast_food' in i else 0 for i in root_listdir])
# randomly choose training and testing cases
permutation = np.random.permutation(len_data)
train_set = [root_listdir[i] for i in permutation[:][:train_examples]]
test_set = [root_listdir[i] for i in permutation[train_examples:-valid_examples]]
valid_set = [root_listdir[i] for i in permutation[-valid_examples:]]
train_labels = labels[permutation[:train_examples]]
test_labels = labels[permutation[train_examples:-valid_examples]]
valid_labels = labels[permutation[-valid_examples:]]
train_folder = root_dir + '/train'
test_folder = root_dir + '/test'
valid_folder = root_dir + '/valid'
os.makedirs(train_folder+'/fast_food/')
os.makedirs(train_folder+'/slow_food/')
os.makedirs(test_folder+'/fast_food/')
os.makedirs(test_folder+'/slow_food/')
os.makedirs(valid_folder+'/fast_food/')
os.makedirs(valid_folder+'/slow_food/')
for f,i in zip(train_set, train_labels):
if i==0:
shutil.move(root_dir + '/' + f, train_folder+'/slow_food/')
else:
shutil.move(root_dir + '/' + f, train_folder+'/fast_food/')
for f,i in zip(test_set, test_labels):
if i==0:
shutil.move(root_dir + '/' + f, test_folder+'/slow_food/')
else:
shutil.move(root_dir + '/' + f, test_folder+'/fast_food/')
for f,i in zip(valid_set, valid_labels):
if i==0:
shutil.move(root_dir + '/' + f, valid_folder+'/slow_food/')
else:
shutil.move(root_dir + '/' + f, valid_folder+'/fast_food/')
Classification models¶
In [13]:
import numpy as np
import tensorflow as tf
import keras
import matplotlib.pyplot as plt
import matplotlib.image as img
import random
import itertools
from PIL import Image
from sklearn.metrics import confusion_matrix
from keras.preprocessing.image import ImageDataGenerator
from keras import applications, backend as K
from keras.models import Sequential, Model
from keras.layers import Dense, Activation, Conv2D, MaxPool2D, Flatten, BatchNormalization, Dropout
from keras.layers.core import Dense, Flatten
from keras.layers.normalization import BatchNormalization
from keras.layers.convolutional import *
from keras.optimizers import Adam
from keras.metrics import categorical_crossentropy
%matplotlib inline
In [14]:
channels = 3
batch_size = 32
Functions¶
In [15]:
def plot_accuracy_loss(history):
"""
Function plotting:
:accuracy and validation accuracy on firs plot
:loss and validation loss on second plot
"""
fig = plt.figure(figsize=(10,5))
plt.subplot(1, 2, 1)
plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('model accuracy')
plt.ylabel('accuracy')
plt.xlabel('epoch')
plt.legend(['train', 'validation'], loc='upper left')
plt.ylim([0, 1])
plt.subplot(1, 2, 2)
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('model loss')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(['train', 'test'], loc='upper right')
plt.show()
In [16]:
def plot_predictions(filenames, preds):
"""
Method plotting test set with predictions.
:param preds - list of predictions for test set.
:param filenames - list of filenames in test set.
"""
root_listdir = filenames[:60]
preds = preds[:60]
fig, ax = plt.subplots(10, 6, frameon=False, figsize=(15, 25))
fig.suptitle('Predictions from test set', fontsize=20)
row = 0
col = 0
wrong_prediction_filenames = []
ec = (0, .6, .1)
for i in range(len(root_listdir)):
try:
if 'slow_food' in root_listdir[int(i)]:
last_dir = 'slow_food'
class_number = 0
else:
last_dir = 'fast_food'
class_number = 1
if preds[i] == 0:
class_name = 'slow food'
else:
class_name = 'fast food'
image = Image.open(root_dir + '/test/' + last_dir + '/' + root_listdir[int(i)])
image = image.resize((120, 120), Image.ANTIALIAS)
ax[row][col].imshow(image)
if class_number != preds[i]:
fc = (1, .0, .0)
wrong_prediction_filenames.append(root_listdir[int(i)])
else:
fc = (0, .7, .2)
ax[row][col].text(0, -20, class_name, size=10, rotation=0, ha='left',
va='top', bbox=dict(boxstyle='round', ec=ec, fc=fc))
ax[row][col].set_xticks([])
ax[row][col].set_yticks([])
col += 1
if col % 6 == 0 and col*row != 54:
col = 0
row += 1
elif col*row == 54:
break
except (OSError, ValueError) as e:
print(e)
return wrong_prediction_filenames
In [17]:
def plot_confusion_matrix(cm, classes, normalize=False, title='Confusion matrix', cmap=plt.cm.Blues):
"""
This function prints and plots the confusion matrix.
Normalization can be applied by setting 'normalize=True'
"""
plt.imshow(cm, interpolation='nearest', cmap=cmap)
plt.title(title)
plt.colorbar()
tick_marks = np.arange(len(classes))
plt.xticks(tick_marks, classes, rotation=45)
plt.yticks(tick_marks, classes)
if normalize:
cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]
print('Normalized confusion matrix')
else:
print('Confusion matrix, without normalization')
print(cm)
tresh = cm.max() / 2
for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
plt.text(round(j, 2), round(i, 2), round(cm[i, j], 2), horizontalalignment='center', color='white' if cm[i, j] > tresh else 'black')
ax = plt.gca() # only to illustrate what `ax` is
ax.autoscale(enable=True, axis='y', tight=False)
plt.tight_layout()
plt.ylabel('True label')
plt.xlabel('Predicted label')
Logistic regression¶
First attempt should be simple so we will check Logistic Regression on our data.
In [18]:
img_height = img_width = 224
batch_size = 32
channels = 3
In [19]:
train_generator = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
train_folder,
color_mode = "rgb",
target_size=(img_height, img_width),
batch_size=batch_size,
class_mode='binary')
test_generator = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
train_folder,
color_mode = "rgb",
target_size=(img_height, img_width),
batch_size=batch_size,
class_mode='binary')
x_train, y_train = next(train_generator)
x_test, y_test = next(test_generator)
In [20]:
from sklearn.linear_model import LogisticRegression
logistic = LogisticRegression(solver='liblinear')
logistic.fit(x_train.reshape(batch_size,-1), y_train)
Out[20]:
In [21]:
y_pred = logistic.predict(x_test.reshape(len(x_test), -1))
y_pred[:10]
Out[21]:
In [22]:
logistic.predict_proba(x_test[:3].reshape(3,-1))
Out[22]:
In [23]:
np.count_nonzero(y_pred == y_test)/len(y_test)
Out[23]:
Model accuracy is not satisfactory.
Build Fine-tuned VGG16 model¶
Now we will try to use pretrained model VGG16. The model achieves 92.7% top-5 test accuracy in ImageNet, which is a dataset of over 14 million images belonging to 1000 classes.
In [24]:
K.clear_session()
In [25]:
vgg16_model = applications.vgg16.VGG16()
In [26]:
vgg16_model = applications.vgg16.VGG16()
In [27]:
vgg16_model.summary()
In [28]:
model = Sequential()
for layer in vgg16_model.layers:
model.add(layer)
In [29]:
model.summary()
We should pop dense layers.
In [30]:
model.layers.pop()
model.layers.pop()
model.layers.pop()
Out[30]:
In [31]:
for layer in model.layers:
layer.trainable = False
In [32]:
model.add(Dense(2, activation='sigmoid'))
In [33]:
model.summary()
In [34]:
img_height = img_width = 224
batch_size = 32
channels = 3
train_batches = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
train_folder,
color_mode = "rgb",
target_size=(img_height, img_width),
batch_size=10,
classes=['fast_food', 'slow_food'])
test_batches = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
test_folder,
color_mode = "rgb",
target_size=(img_height, img_width),
batch_size=10,
classes=['fast_food', 'slow_food'],
shuffle=False)
valid_batches = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
valid_folder,
color_mode = "rgb",
target_size=(img_height, img_width),
batch_size=4,
classes=['fast_food', 'slow_food'])
Train the fine-tuned VGG16 model¶
In [35]:
model.compile(Adam(lr=.001), loss='categorical_crossentropy', metrics=['accuracy'])
In [36]:
history = model.fit_generator(train_batches, train_examples//10, validation_data=valid_batches, validation_steps= valid_examples//4 + 1, epochs=4, verbose=2)
In [37]:
predictions = model.predict_generator(test_batches, test_examples//10)
In [38]:
correct = 0
for i, f in enumerate(test_batches.filenames[:len(predictions)]):
if 'slow_food' in f and predictions[i][0]<0.5:
correct +=1
if 'fast_food' in f and predictions[i][0]>=0.5:
correct +=1
print('Correct predictions: '+str(correct/len(test_batches.filenames)))
In [39]:
plot_accuracy_loss(history)
In [40]:
c = list(zip([i.split('\\')[1] for i in test_batches.filenames],[1 if i[0] >= 0.5 else 0 for i in predictions]))
my_list = random.shuffle(c)
filenames, preds = zip(*c)
In [41]:
wrong_predictions = plot_predictions(filenames, preds)
In [42]:
wrong_predictions
Out[42]:
In [43]:
test_labels = np.array([0 if 'slow_food' in f else 1 for f in test_batches.filenames])[:len(predictions)]
cm = confusion_matrix(test_labels, np.round(predictions[:,0]))
In [44]:
cm_plot_labels = ['fast_food', 'slow_food']
plot_confusion_matrix(cm, cm_plot_labels, title='Confusion Matrix')
Model performed better than simple guessing, but it is still less accurate than Logistic Regression.
Sequential model¶
Pretrained model didn't pass the exam. Now we will try create our own model.
In [45]:
K.clear_session()
In [46]:
model = Sequential()
model.add(Conv2D(8, kernel_size=(3,3), padding='same', input_shape = (img_width,img_height,channels)))
model.add(Activation('relu'))
model.add(MaxPool2D(pool_size=(2, 2)))
model.add(Conv2D(16, kernel_size=(3,3), padding='same'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPool2D(pool_size=(2, 2)))
model.add(Conv2D(32, kernel_size=(3,3), padding='same'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPool2D(pool_size=(2, 2)))
model.add(Flatten())
model.add(Dense(2, activation='sigmoid'))
model.compile(optimizer='rmsprop', loss='binary_crossentropy', metrics=['accuracy'])
In [47]:
model.summary()
In [48]:
train_batches= ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
train_folder,
color_mode = "rgb",
target_size=(img_height, img_width),
batch_size=batch_size,
classes=['fast_food', 'slow_food'])
test_batches = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
test_folder,
color_mode = "rgb",
target_size=(img_height, img_width),
batch_size=batch_size,
classes=['fast_food', 'slow_food'],
shuffle=False)
In [49]:
model.fit_generator(train_batches, train_examples//batch_size, epochs=2)
Out[49]:
In [50]:
y_pred = model.predict_generator(test_batches, test_examples//batch_size, workers=4)
In [51]:
correct = 0
for i, f in enumerate(test_batches.filenames[:len(y_pred)]):
if 'slow_food' in f and y_pred[i][0]<0.5:
correct +=1
if 'fast_food' in f and y_pred[i][0]>=0.5:
correct +=1
print('Correct predictions: '+str(correct/len(test_batches.filenames)))
In [52]:
c = list(zip([i.split('\\')[1] for i in test_batches.filenames],[1 if i[0] >= 0.5 else 0 for i in y_pred]))
my_list = random.shuffle(c)
filenames_2, preds_2 = zip(*c)
In [53]:
wrong_predictions_2 = plot_predictions(filenames_2, preds_2)
In [54]:
wrong_predictions_2
Out[54]:
In [55]:
test_labels = np.array([0 if 'slow_food' in f else 1 for f in test_batches.filenames])[:len(y_pred)]
cm = confusion_matrix(test_labels, np.round(y_pred[:, 0]))
In [56]:
cm_plot_labels = ['fast_food', 'slow_food']
plot_confusion_matrix(cm, cm_plot_labels, title='Confusion Matrix')
Model is much worse than previous ones. It means that simple model won't be enough, because we don't have big dataset.
Transfer learning¶
Last attempt is transfer learning. We will use VGG16 again but this time in other way.
In [57]:
K.clear_session()
In [58]:
model = applications.VGG16(include_top=False, input_shape=(img_width, img_height, channels))
In [59]:
batch_size = 32
img_height = img_width = 224
channels = 3
train_batches_1 = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
train_folder,
target_size=(img_height, img_width),
batch_size=batch_size,
class_mode=None,
shuffle=False)
train_batches_2 = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
train_folder,
target_size=(img_height, img_width),
batch_size=batch_size,
class_mode=None,
shuffle=False)
test_batches = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
test_folder,
target_size=(img_height, img_width),
batch_size=batch_size,
class_mode='binary',
shuffle=False)
valid_batches = ImageDataGenerator(rescale=1./255,
rotation_range=5,
zoom_range=0.2,
horizontal_flip=True).flow_from_directory(
valid_folder,
target_size=(img_height, img_width),
batch_size=batch_size,
class_mode=None,
shuffle=False)
In [60]:
model = applications.VGG16(include_top=False, weights='imagenet')
model.summary()
Training VGG16 on our data should give us the bottleneks of our images. In this case we will use VGG16 predictions to train our sequential model. It means that we changed sets of images into sets of bottlenecks which could be more helpful in detecting category.
In [61]:
bottleneck_features_train = model.predict_generator(train_batches_1, train_examples//batch_size, verbose=1, workers=4)
bottleneck_features_valid = model.predict_generator(valid_batches, test_examples//batch_size, verbose=1, workers=4)
bottleneck_features_test = model.predict_generator(test_batches, test_examples//batch_size, verbose=1, workers=4)
bottleneck_features_train.shape
Out[61]:
In [62]:
model = Sequential()
model.add(Flatten(input_shape=bottleneck_features_train.shape[1:]))
model.add(Dense(256, activation='relu'))
model.add(Dropout(0.5))
model.add(Dense(1, activation='sigmoid'))
model.compile(optimizer='rmsprop',
loss='binary_crossentropy', metrics=['accuracy'])
model.summary()
In [63]:
labels = np.array([0 if 'slow_food' in f else 1 for f in train_batches_2.filenames])[:len(bottleneck_features_train)]
val_labels = np.array([0 if 'slow_food' in f else 1 for f in valid_batches.filenames])[:len(bottleneck_features_valid)]
history_2 = model.fit(bottleneck_features_train, labels, validation_data=(bottleneck_features_valid[:len(val_labels)], val_labels), epochs=20, batch_size=batch_size)
In [64]:
test_labels = np.array([0 if 'slow_food' in f else 1 for f in test_batches.filenames])[:len(bottleneck_features_test)]
y_test_pred = model.predict_classes(bottleneck_features_test)
accuracy = np.count_nonzero(y_test_pred[:len(test_labels)].ravel() == test_labels)/len(test_labels)
print('\nThe accuracy is: '+str(accuracy))
In [65]:
correct = 0
for i, f in enumerate(test_batches.filenames[:len(test_labels)]):
if 'slow_food' in f and y_test_pred[i] == 0:
correct +=1
if 'fast_food' in f and y_test_pred[i] == 1:
correct +=1
print('Correct predictions: '+str(correct/len(test_batches.filenames[:len(test_labels)])))
In [66]:
plot_accuracy_loss(history_2)
This model is the most promising one.
In [67]:
cm = confusion_matrix(test_labels, np.round(y_test_pred[:len(test_labels)]))
cm_plot_labels = ['slow_food', 'fast_food']
In [68]:
plot_confusion_matrix(cm, cm_plot_labels, title='Confusion Matrix')
In [69]:
plot_confusion_matrix(cm, cm_plot_labels, title='Confusion Matrix', normalize=True)
Prediction on the test set are also much more accurate thane previous ones.
In [70]:
c = list(zip([i.split('\\')[1] for i in test_batches.filenames], y_test_pred))
my_list = random.shuffle(c)
filenames_3, preds_3 = zip(*c)
In [71]:
wrong_predictions_3 = plot_predictions(filenames_3, preds_3)
In [72]:
wrong_predictions_3
Out[72]: